package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.utils.HBaseUtil;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @author gufg
 * @since 2025-09-22 09:50
 */
public class HBaseSinkFunc extends RichSinkFunction<String> {

    private String namespace;

    private byte[][] splitKeys = new byte[7][];

    public HBaseSinkFunc(String namespace, byte[][] splitKeys) {
        this.namespace = namespace;

        for (int i = 0; i < splitKeys.length; i++) {
            this.splitKeys[i] = splitKeys[i];
        }
    }

    @Override
    public void invoke(String jsonStr, Context context) throws Exception {
        // {"rowkey":"511603","sink_family":"info","after":{"code":"511603","name":"四川省广安市前锋区"},"table":"dim_area_info"}
        JSONObject jsonObj = JSON.parseObject(jsonStr);

        // 表名
        String table = jsonObj.getString("table");
        String familyName = jsonObj.getString("sink_family");
        // 实现逻辑
        // 判断表是否，不存创建
        HBaseUtil.createTable(namespace, table, familyName, splitKeys);

        // 插入数据到HBase
        HBaseUtil.put(namespace, table, jsonObj);
    }
}
